package com.wsjj.yjh;

import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;


public class OperatorBroadcastStateDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

//        env.setStateBackend(new HashMapStateBackend());

        env.setStateBackend(new EmbeddedRocksDBStateBackend());

        System.setProperty("HADOOP_USER_NAME","atguigu");

//        1.启动检查点：默认是borrier对齐的，周期为5s，精确一次
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
//        2.指定检查点的存储位置
        checkpointConfig.setCheckpointStorage("htfs://hadoop102:8020/chk");
//        3.checkpoint的超时时间：默认10分钟
        checkpointConfig.setCheckpointTimeout(60000);
//        4.同时运行中的checkpoint的最大数量,默认是1，一般都是1
        checkpointConfig.setMaxConcurrentCheckpoints(1);
//        5.最小等待间隔：上一轮checkpoint结束  到  下一轮checkpoint开始  之间的间隔，设置>0,并发就会变成1
        checkpointConfig.setMinPauseBetweenCheckpoints(1000);
//        6.取消作业时，checkpoint的数据，是否保留在外部系统
//          DELETE_ON_CANCELLATION:动cancel时，删除存在外部系统的chk-xx目录(如果是程序笑然挂掉，不会删)
        checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        7.允许checkpint  连续失败的次数，默认 0---> 表示checkpoint一失败，job就挂掉
        checkpointConfig.setTolerableCheckpointFailureNumber(10);

//        TODO 开启  非对齐检查点（barrier非对齐）
//        开启的要求：checkpoint模式必须是精确一次，最大并发必须设为1
        checkpointConfig.enableUnalignedCheckpoints();
//        开启非对齐检查点才生效：  默认0， 表示一开始就直接用   非对齐的检查点
//        如果大于0，一开始用  对齐的检查点（barrier对齐），对齐的时间超过这个参数，自动切换成   非对齐检查点  （barrier非对齐）
        checkpointConfig.setAlignedCheckpointTimeout(Duration.ofSeconds(1));







        SingleOutputStreamOperator<WaterSensor> DataDS = env.socketTextStream("1.94.41.70", 7777)
                .map(value -> {
                    String[] split = value.split(",");
                    return new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2]));
                });

        DataStreamSource<String> ConfigDS = env.socketTextStream("1.94.41.70", 8888);
        MapStateDescriptor<String, Integer> broad = new MapStateDescriptor<>("broad", Types.STRING, Types.INT);
        BroadcastStream<String> broadcast = ConfigDS.broadcast(broad);

        BroadcastConnectedStream<WaterSensor, String> connect = DataDS.connect(broadcast);
        connect.process(new BroadcastProcessFunction<WaterSensor, String, String>() {
            @Override
            public void processElement(WaterSensor value, BroadcastProcessFunction<WaterSensor, String, String>.ReadOnlyContext ctx, Collector<String> out) throws Exception {
                ReadOnlyBroadcastState<String, Integer> broadcastState = ctx.getBroadcastState(broad);
                Integer broad1 = broadcastState.get("broad");
                Integer vc = value.getVc();
                int broad2 = broad1 == null ? 0 : broad1;
                if (vc > broad2){
                    out.collect(value + ",水位超过指定的阈值：" + broad2 + "!!!");
                }

            }

            @Override
            public void processBroadcastElement(String value, BroadcastProcessFunction<WaterSensor, String, String>.Context ctx, Collector<String> out) throws Exception {
                BroadcastState<String, Integer> broadcastState = ctx.getBroadcastState(broad);
                broadcastState.put("broad",Integer.valueOf(value));

            }
        })

                .print();
        env.execute();


    }
}
